package org.netwia.networks;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.http.HttpEntity;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.mime.MultipartEntityBuilder;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.util.EntityUtils;
import org.netwia.utils.StartStopService;
import org.netwia.utils.ValidationUtils;

/**
 * This is implementation of remote network connector which works as http client.
 * This periodically sends requests to the server. Request and response contain the queued data.
 * 
 * @author radek.hecl
 *
 * @param <T> type of the network parameter
 */
public class HttpRemoteNetworkConnectorClient<T> implements RemoteNetworkConnector<T>, StartStopService {

    /**
     * Builder class.
     * 
     * @param <T> type of the network parameter
     *
     */
    public static class Builder<T> {

        /**
         * URL to connect with.
         */
        private String url;

        /**
         * Client name of this client. This is the unique sender identification.
         */
        private String localClientName;

        /**
         * Streamer for the packages.
         */
        private RemotePackageStreamer<T> streamer;

        /**
         * Delay between requests.
         */
        private Integer delay;

        /**
         * Sets URL to connect with.
         * 
         * @param url URL to connect with
         * @return this instance
         */
        public Builder<T> setUrl(String url) {
            this.url = url;
            return this;
        }

        /**
         * Sets local client name. This is the unique client identification
         * 
         * @param localClientName local client name
         * @return this instance
         */
        public Builder<T> setLocalClientName(String localClientName) {
            this.localClientName = localClientName;
            return this;
        }

        /**
         * Sets streamer for packages.
         * 
         * @param streamer streamer
         * @return this instance
         */
        public Builder<T> setStreamer(RemotePackageStreamer<T> streamer) {
            this.streamer = streamer;
            return this;
        }

        /**
         * Sets delay between requests.
         *  
         * @param delay delay between requests
         * @return this instance
         */
        public Builder<T> setDelay(int delay) {
            this.delay = delay;
            return this;
        }

        /**
         * Builds the result object.
         * 
         * @return created object
         */
        public HttpRemoteNetworkConnectorClient<T> build() {
            return new HttpRemoteNetworkConnectorClient<T>(this);
        }
    }

    /**
     * URL to connect with.
     */
    private String url;

    /**
     * Client name of this client. This is the unique sender identification.
     */
    private String localClientName;

    /**
     * Streamer for the packages.
     */
    private RemotePackageStreamer<T> streamer;

    /**
     * Delay between requests.
     */
    private Integer delay;

    /**
     * Outputs ready to be sent.
     */
    private Queue<RemotePackage<T>> outputs = new LinkedBlockingQueue<RemotePackage<T>>();

    /**
     * Map with inputs ready for this network.
     */
    private Map<String, Queue<T>> inputs = new ConcurrentHashMap<String, Queue<T>>();

    /**
     * Lock object.
     */
    private Lock lock = new ReentrantLock();

    /**
     * Flag indicating whether processing is running or not.
     */
    private boolean running = false;

    /**
     * Thread in which the processing is running.
     */
    private Thread processingThread = null;

    /**
     * Creates new instance.
     * 
     * @param builder builder object
     */
    public HttpRemoteNetworkConnectorClient(Builder<T> builder) {
        url = builder.url;
        localClientName = builder.localClientName;
        streamer = builder.streamer;
        delay = builder.delay;
        guardInvariants();
    }

    /**
     * Guards this object to be consistent. Throws exception if this is not the case.
     */
    private void guardInvariants() {
        ValidationUtils.guardNotEmpty(url, "url cannot be empty");
        ValidationUtils.guardNotEmpty(localClientName, "localClientName cannot be empty");
        ValidationUtils.guardNotNull(streamer, "streamer cannot be null");
        ValidationUtils.guardNotNegativeInt(delay, "dalay cannot be negative");
    }

    @Override
    public void sendOutput(String targetClient, String packageName, T data) {
        outputs.add(new RemotePackage.Builder<T>().
                setSenderClient(localClientName).
                setTargetClient(targetClient).
                setPackageName(packageName).
                setData(data).
                build());
    }

    @Override
    public boolean isNextInputReady(String packageName) {
        Queue<T> q = inputs.get(packageName);
        if (q == null) {
            return false;
        }
        return !q.isEmpty();
    }

    @Override
    public T getNextInput(String packageName) {
        Queue<T> q = inputs.get(packageName);
        if (q == null) {
            throw new NoSuchElementException("no element on the input, even no queue: packageName = " + packageName);
        }
        T res = q.poll();
        if (res == null) {
            throw new NoSuchElementException("no element on the input: packageName = " + packageName);
        }
        return res;
    }

    @Override
    public void start() {
        lock.lock();
        try {
            if (running) {
                return;
            }
            processingThread = new Thread(new HttpClientRunnable<T>(this));
            processingThread.start();
            running = true;
        } finally {
            lock.unlock();
        }
    }

    @Override
    public void stop() {
        lock.lock();
        try {
            if (!running) {
                return;
            }
            running = false;
            processingThread.join();
            processingThread = null;
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            lock.unlock();
        }
    }

    @Override
    public String toString() {
        return ToStringBuilder.reflectionToString(this);
    }

    /**
     * Inner class invokable in the separate thread.
     * 
     * @param <T> type of the network data 
     */
    private static class HttpClientRunnable<T> implements Runnable {

        /**
         * Owner of this class.
         */
        private HttpRemoteNetworkConnectorClient<T> owner;

        /**
         * Creates new instance.
         * 
         * @param owner owner of this class
         */
        public HttpClientRunnable(HttpRemoteNetworkConnectorClient<T> owner) {
            this.owner = owner;
            guardInvariants();
        }

        /**
         * Guards this object to be consistent. Throws exception if this is not the case.
         */
        private void guardInvariants() {
            ValidationUtils.guardNotNull(owner, "owner cannot be null");
        }

        @SuppressWarnings("resource")
        @Override
        public void run() {
            CloseableHttpClient client = null;
            try {
                client = HttpClientBuilder.create().
                        build();
                while (owner.running) {
                    try {
                        HttpPost httpPost = new HttpPost(owner.url);

                        MultipartEntityBuilder reqEntityBuilder = MultipartEntityBuilder.create().
                                addTextBody("senderClient", owner.localClientName);
                        List<RemotePackage<T>> packages = new ArrayList<RemotePackage<T>>();
                        while (!owner.outputs.isEmpty()) {
                            packages.add(owner.outputs.poll());
                        }
                        for (int i = 0; i < packages.size(); ++i) {
                            RemotePackage<T> pckg = packages.get(i);
                            reqEntityBuilder.addTextBody("package" + i + ".targetClient", pckg.getTargetClient());
                            ByteArrayOutputStream bos = null;
                            try {
                                bos = new ByteArrayOutputStream();
                                owner.streamer.writeToStream(pckg, bos);
                                bos.close();
                            } finally {
                                IOUtils.closeQuietly(bos);
                            }
                            reqEntityBuilder.addBinaryBody("package" + i + ".bin", bos.toByteArray());
                        }

                        HttpEntity reqEntity = reqEntityBuilder.build();

                        httpPost.setEntity(reqEntity);

                        CloseableHttpResponse response = null;
                        try {
                            response = client.execute(httpPost);
                            if (response.getEntity().getContentLength() > 0) {
                                InputStream is = null;
                                try {
                                    // read to byte array
                                    byte[] bytes = EntityUtils.toByteArray(response.getEntity());
                                    
                                    // convert to objects
                                    is = new ByteArrayInputStream(bytes);
                                    while (is.available() > 0) {
                                        RemotePackage<T> respPckg = owner.streamer.readFromStream(is);
                                        if (!owner.inputs.containsKey(respPckg.getPackageName())) {
                                            owner.inputs.put(respPckg.getPackageName(), new LinkedBlockingQueue<T>());
                                        }
                                        owner.inputs.get(respPckg.getPackageName()).add(respPckg.getData());
                                    }
                                    is.close();
                                } finally {
                                    IOUtils.closeQuietly(is);
                                }
                            }
                            response.close();
                        } finally {
                            IOUtils.closeQuietly(response);
                        }
                    } catch (UnsupportedEncodingException e) {
                        throw new RuntimeException(e);
                    } catch (ClientProtocolException e) {
                        throw new RuntimeException(e);
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }

                    if (owner.delay > 0) {
                        try {
                            Thread.sleep(owner.delay);
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                    }
                }

                client.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            } finally {
                IOUtils.closeQuietly(client);
            }
        }

        @Override
        public String toString() {
            return ToStringBuilder.reflectionToString(this);
        }
    }

}
